package app

import com.cw.realtime.common.base.FlinkRunner.{RunnerConfig, run}
import com.cw.realtime.common.bean.TrafficPageViewBean
import com.cw.realtime.common.constant.Constant._
import com.cw.realtime.common.util.{DateFormatUtil, FlinkSinkUtil}
import io.circe._
import io.circe.parser._
import io.circe.syntax._
import org.apache.flink.api.common.eventtime.{SerializableTimestampAssigner, WatermarkStrategy}
import org.apache.flink.api.common.state.{StateTtlConfig, ValueState, ValueStateDescriptor}
import org.apache.flink.configuration.Configuration
import org.apache.flink.streaming.api.functions.KeyedProcessFunction
import org.apache.flink.streaming.api.scala.function.ProcessWindowFunction
import org.apache.flink.streaming.api.scala.{DataStream, createTypeInformation}
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows
import org.apache.flink.streaming.api.windowing.time.Time
import org.apache.flink.streaming.api.windowing.windows.TimeWindow
import org.apache.flink.util.Collector

import java.time.Duration

object DwsTrafficUserVisitPageView {

  def main(args: Array[String]): Unit = {
    implicit val config = RunnerConfig("traffic-user-visit-page-view", TOPIC_DWD_TRAFFIC_PAGE, 10022, ckp = true)
    run { (env, ds) =>
      //      ds.print()

      // 数据清洗
      val jsonObjectStream = etl(ds)

      // 按照mid分组，查找状态判断当前记录是否是独立访客，最后输出成PageView对象，封装统计信息
      val pageViewStream = getPageView(jsonObjectStream)

      // 添加水位线
      val pageViewStreamWM = addWatermark(pageViewStream)

      // 按照多粒度分组, 然后开窗聚合
      val aggStream = groupedWindowAgg(pageViewStreamWM)

      // 写入到doris
      write2Doris(aggStream)


    }

  }

  def etl(ds: DataStream[String]) = {
    ds.flatMap { s =>
      for {
        json <- parse(s).toOption
        jsonObj <- json.asObject
        common <- jsonObj("common")
        common <- common.asObject
        mid <- common("mid") if !mid.isNull // 防止后续mid分组报错
      } yield jsonObj
    }
  }


  def getPageView(jsonObjectStream: DataStream[JsonObject]) = {
    jsonObjectStream
      .keyBy {
        _("common").get.asObject.get("mid").get.asString.get
      }
      .process(new KeyedProcessFunction[String, JsonObject, TrafficPageViewBean] {
        type FuncType = KeyedProcessFunction[String, JsonObject, TrafficPageViewBean]
        val stateDesc = new ValueStateDescriptor[String]("last_login", classOf[String])
        var loginState = Option.empty[ValueState[String]]

        override def open(parameters: Configuration): Unit = {
          stateDesc.enableTimeToLive(StateTtlConfig.newBuilder(org.apache.flink.api.common.time.Time.days(1)).build())
          loginState = Some(getRuntimeContext.getState(stateDesc))

        }

        override def processElement(value: JsonObject, ctx: FuncType#Context, out: Collector[TrafficPageViewBean]): Unit = {
          for {
            ts <- value("ts").get.asNumber.get.toLong
            common <- value("common").get.asObject
            sid = common("sid").getOrElse("".asJson).asString.get
            page <- value("page").get.asObject
            ar <- common("ar").get.asString
            ch <- common("ch").get.asString
            vc <- common("vc").get.asString
            isNew <- common("is_new").get.asString
            dur <- page("during_time").get.asNumber.get.toLong
            state <- loginState
            curLoginDt = DateFormatUtil.tsToDate(ts)
            lastLoginDt = state.value
          } {
            // 判断独立访客
            val userVisit =
              if (lastLoginDt == null || curLoginDt != lastLoginDt) {
                state.update(curLoginDt);
                1
              } else 0


            // 判断新会话
            val lastPageId = page("last_page_id")
            val sessionCnt =
              if (lastPageId.isEmpty || lastPageId.get.isNull) 1 else 0

            out.collect(TrafficPageViewBean(vc, ch, ar, isNew, userVisit, sessionCnt, 1, dur, ts, sid))
          }
        }
      })
  }

  def addWatermark(ds: DataStream[TrafficPageViewBean]) = {
    ds.assignTimestampsAndWatermarks(WatermarkStrategy.forBoundedOutOfOrderness(Duration.ofSeconds(3))
      .withTimestampAssigner(new SerializableTimestampAssigner[TrafficPageViewBean] {
        override def extractTimestamp(element: TrafficPageViewBean, recordTimestamp: Long): Long =
          element.ts
      }))
  }

  def groupedWindowAgg(ds: DataStream[TrafficPageViewBean]) = {
    ds.keyBy(bean => s"${bean.vc}:${bean.ch}:${bean.ar}")
      .window(TumblingEventTimeWindows.of(Time.seconds(10)))
      .reduce(_ + _, new ProcessWindowFunction[TrafficPageViewBean, TrafficPageViewBean, String, TimeWindow] {
        override def process(key: String, context: Context, elements: Iterable[TrafficPageViewBean], out: Collector[TrafficPageViewBean]): Unit = {
          val win = context.window
          val stt = DateFormatUtil.tsToDateTime(win.getStart)
          val edt = DateFormatUtil.tsToDateTime(win.getEnd)
          val dt = DateFormatUtil.tsToDateForPartition(System.currentTimeMillis())
          elements.foreach { elem => out.collect(elem.transWindowRange(stt, edt, dt)) }
        }
      })
  }

  def write2Doris(ds: DataStream[TrafficPageViewBean]) = {
    ds.map { view =>
      import io.circe.generic.auto._
      val notSerialized = Seq("ts", "sid")
      view.asJson
        .asObject.get
        .filterKeys(!notSerialized.contains(_))
        .toJson.noSpaces
    }.sinkTo(FlinkSinkUtil.getDorisSink(DWS_TRAFFIC_USER_VISIT_PAGE_VIEW_WINDOW))

  }

}
